package tqueue

import (
	"fmt"
	"sync/atomic"
	"time"
	"unsafe"
)

// private structure
type node struct {
	value interface{}
	next  *node
}

type Queue struct {
	dummy *node
	tail  *node
	count int32

	notify chan struct{}
}

func NewQueue() *Queue {
	q := new(Queue)
	q.dummy = new(node)
	q.tail = q.dummy

	q.notify = make(chan struct{}, 100)

	return q
}

func (q *Queue) Enqueue(v interface{}) {
	var oldTail, oldTailNext *node

	newNode := new(node)
	newNode.value = v

	newNodeAdded := false

	for !newNodeAdded {
		oldTail = q.tail
		oldTailNext = oldTail.next

		if q.tail != oldTail {
			continue
		}

		if oldTailNext != nil {
			atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&q.tail)), unsafe.Pointer(oldTail), unsafe.Pointer(oldTailNext))
			continue
		}

		newNodeAdded = atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&oldTail.next)), unsafe.Pointer(oldTailNext), unsafe.Pointer(newNode))
	}

	atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&q.tail)), unsafe.Pointer(oldTail), unsafe.Pointer(newNode))
	atomic.AddInt32(&(q.count), 1)
	if len(q.notify) < 100 {
		go func() {
			q.notify <- struct{}{}
		}()
	}
}

func (q *Queue) Close() {
	close(q.notify)
}

func (q *Queue) Dequeue() (interface{}, bool) {
	var temp interface{}
	var oldDummy, oldHead *node

	removed := false

	for !removed {
		oldDummy = q.dummy
		oldHead = oldDummy.next
		oldTail := q.tail

		if q.dummy != oldDummy {
			continue
		}

		if oldHead == nil {
			return nil, false
		}

		if oldTail == oldDummy {
			atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&q.tail)), unsafe.Pointer(oldTail), unsafe.Pointer(oldHead))
			continue
		}

		temp = oldHead.value
		removed = atomic.CompareAndSwapPointer((*unsafe.Pointer)(unsafe.Pointer(&q.dummy)), unsafe.Pointer(oldDummy), unsafe.Pointer(oldHead))
		if removed {
			atomic.AddInt32(&(q.count), -1)
			//fmt.Println("-1剩余:", atomic.LoadInt32(&(q.count)))
		}
	}

	return temp, true
}

func (q *Queue) iterate(c chan<- interface{}) {
	//fmt.Println(q.notify, len(q.notify))
FOR:
	for {
		select {
		case _, okN := <-q.notify:
			//通知已关闭
			if !okN {
				break FOR
			}

			//消费消息
			for {
				datum, ok := q.Dequeue()
				if !ok {
					fmt.Println("item count:", q.Count())
					break
				}

				c <- datum

			}
		default:
			time.Sleep(time.Millisecond * 200)
		}

	}
	fmt.Println("iterate closed")
	close(c)
}

func (q *Queue) NewIter() <-chan interface{} {
	c := make(chan interface{})
	go q.iterate(c)
	return c
}

func (q *Queue) Count() int {
	return int(atomic.LoadInt32(&(q.count)))
}
